package cn.com.sparkle.paxos.userserver;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.ConfigNode;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.addprocess.AddRequestDealer;
import cn.com.sparkle.paxos.addprocess.AddRequestPackage;
import cn.com.sparkle.paxos.client.MasterMayBeLostException;
import cn.com.sparkle.paxos.client.PaxosClient;
import cn.com.sparkle.paxos.client.PaxosOperater;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.deamon.ReConnectThread.ReConnectMethod;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.MasterChangePosEvent;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CustomOnlyConnectMaster;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MasterBeatHeartRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.raptor.core.buff.SyncBuffPool;
import cn.com.sparkle.raptor.core.handler.IoHandler;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.Protocol;
import cn.com.sparkle.raptor.core.protocol.ProtocolHandler;
import cn.com.sparkle.raptor.core.transport.socket.nio.IoSession;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketClient;
import cn.com.sparkle.raptor.core.transport.socket.nio.exception.SessionHavaClosedException;

public class UserServerHandler implements ProtocolHandler {

	private final static Logger logger = Logger
			.getLogger(UserServerHandler.class);
	private volatile boolean isMaster = false;
	private volatile String masterAddress;
	private ConcurrentHashMap<ProtocolHandlerIoSession, ProtocolHandlerIoSession> wantConnectMasterSessionMap = new ConcurrentHashMap<ProtocolHandlerIoSession, ProtocolHandlerIoSession>();
	private ReentrantLock masterChangelock = new ReentrantLock();
	private UserHandlerInterface handlerInterface;
	private volatile PaxosClient client = null;

	public UserServerHandler(EventsManager eventsManager,
			final Configuration conf, UserHandlerInterface handlerInterface) {
		this.handlerInterface = handlerInterface;
		eventsManager.registerEvent(new MasterChangePosEvent() {
			@Override
			public void masterChange(String address) {
				// nothing to do
				try {
					masterChangelock.lock();
					logger.debug("master changed");
					for (ConfigNode cnode : conf.getSenators()) {
						if ((cnode.getIp() + ":" + cnode.getPort())
								.equals(address)) {
							masterAddress = cnode.getIp() + ":"
									+ cnode.getClientPort();
						}
					}

					if (client == null) {
						try {
							client = new PaxosClient(
									new String[] { masterAddress }, conf
											.getTransportTcpNum(), true);
						} catch (Exception e) {
							logger.error("fatal error", e);
						}
					} else {
						logger.debug("change address");
						try {
							client.changeSenator(new String[] { masterAddress });
						} catch (Exception e) {
							logger.error("fatal error", e);
						}
					}
				} finally {
					masterChangelock.unlock();
				}
			}

			@Override
			public void lostPos() {
				try {
					masterChangelock.lock();
					isMaster = false;
					for (ProtocolHandlerIoSession session : wantConnectMasterSessionMap
							.keySet()) {
						session.closeSession();
					}
				} finally {
					masterChangelock.unlock();
				}
			}

			@Override
			public void getMasterPos() {
				try {
					masterChangelock.lock();
					isMaster = true;
				} finally {
					masterChangelock.unlock();
				}
			}
		});
	}

	@Override
	public void onOneThreadSessionOpen(ProtocolHandlerIoSession session) {
		handlerInterface.onClientConnect(session);
	}

	@Override
	public void onOneThreadSessionClose(ProtocolHandlerIoSession session) {
		wantConnectMasterSessionMap.remove(session);
		handlerInterface.onClientClose(session);
	}

	@Override
	public void onOneThreadCatchException(IoSession session,
			ProtocolHandlerIoSession attachment, Throwable e) {
		logger.error("", e);
	}

	@Override
	public void onOneThreadMessageRecieved(Object receiveObject,
			final ProtocolHandlerIoSession session) {
		if (receiveObject instanceof CustomOnlyConnectMaster) {
			try {
				masterChangelock.lock();
				if (isMaster) {
					wantConnectMasterSessionMap.put(session, session);
				} else {
					session.closeSession();
				}
			} finally {
				masterChangelock.unlock();
			}
		} else if (receiveObject instanceof MessagePackage) {
			MessagePackage messagePackage = (MessagePackage) receiveObject;
			final long packageId = messagePackage.getId();
			if (messagePackage.hasAddRequest()) {

				AddRequest addRequest = messagePackage.getAddRequest();
				if (addRequest.getTransportMaster()
						|| addRequest.getIsManagerCommand()) {
					if (isMaster) {
						handlerInterface.onClientCustomCommandReceive(session,
								messagePackage);
					} else {
						if (client != null) {
							try {
								PaxosOperater operator = (PaxosOperater) session.customAttachment;
								if (operator == null) {
									operator = client.getOperator();
									session.customAttachment = operator;// assign
																		// a
																		// channel
																		// to
																		// master
								}
								try {
									operator.add(addRequest.getValue()
											.toByteArray(), 0, true,
											new PaxosOperater.CallBack() {
												@Override
												public void callBack(
														AddResponse response) {
													try {
														MessagePackage.Builder mp = MessagePackage
																.newBuilder()
																.setIsLast(true)
																.setId(packageId);
														mp.setAddResponse(response);
														session.writeObject(mp
																.build());
													} catch (SessionHavaClosedException e) {
													}
												}
											});
								} catch (MasterMayBeLostException e) {
									throw new RuntimeException("the master may be lost!");
								}
							} catch (InterruptedException e) {
								logger.error("fatal error", e);
								;
							}
						} else {
							session.closeSession();
						}
					}
				} else {
					handlerInterface.onClientCustomCommandReceive(session,
							messagePackage);
				}
			} else if (messagePackage.hasMasterBeatHeartRequest()) {
				// send master beatheart request
				MessagePackage.Builder mp = MessagePackage
						.newBuilder()
						.setIsLast(true)
						.setId(packageId)
						.setMasterBeatHeartRequest(
								MasterBeatHeartRequest.getDefaultInstance());
				try {
					session.writeObject(mp.build());
				} catch (SessionHavaClosedException e) {
					// nothing to do
				}
			} else {
				logger.error("unexcepted message");
			}
		}
	}

	@Override
	public void onOneThreadMessageSent(ProtocolHandlerIoSession attachment,
			int sendSize) {
		// nothing to do
	}

}
